//
// Created by 郝羽 on 2018/6/25.
//

#include "master.hpp"

#include <glib.h>
#include <glib-unix.h>
#include <set>
#include <vector>

#include "clock.hpp"
#include "conn.hpp"
#include "config.hpp"
#include "metric_server.hpp"
#include "subprocess.hpp"
#include "report.hpp"

constexpr std::chrono::milliseconds kMasterTimeoutInterval(20);

struct Master {
    std::set<subprocess*> workers;
    Context ctx;
    MetricServer* mserver = nullptr;
    GMainLoop* loop = nullptr;
    std::vector<Conn*> conns;
    Clock::time_point start_time;
    bool exit = false;
};

gboolean master_handle_signal(gpointer user_data) {
    auto m = static_cast<Master*>(user_data);

    g_message("master received signal");
    for (auto worker : m->workers) {
        subprocess_signal(worker, SIGINT);
    }

    return G_SOURCE_REMOVE;
}

void master_handle_child_exit(subprocess *child, void *data) {
    auto m = static_cast<Master*>(data);
    m->workers.erase(child);
    subprocess_free(child);
    if (m->workers.empty()) {
        //g_message("stop loop");
        //g_main_loop_quit(m->main_loop);
        m->exit = true;
    }
}

void master_handle_conn_init(Conn* conn, bool ok, void* data) {
    auto m = static_cast<Master*>(data);

    if (!ok) {
        return;
    }

    m->ctx.pub_rtc_ids.emplace_back(conn->remote_rtc_id().c_str());

    if (m->ctx.sequence == Context::Sequence::Offerer) {
        conn->trickle();
    } else if (m->ctx.sequence == Context::Sequence::Answerer) {
        conn->answer();
    }
}

void master_handle_conn_subscribe(Conn* conn, bool ok, void* data) {
    if (!ok) {
        conn->terminate();
        return;
    }
}

void master_handle_conn_trickle(Conn* conn, bool ok, void* data) {
    if (!ok) {
        conn->terminate();
        return;
    }
}

void master_handle_conn_terminate(Conn* conn, bool ok, void* data) {
    // nothing to do
}

void master_handle_conn_answer(Conn* conn, bool ok, void* data) {
    if (!ok) {
        conn->terminate();
        return;
    }

    conn->trickle();
}

void master_handle_conn_setup(Conn* conn, int setup_time, void* data) {

}

ConnectionEventHandler master_pub_handler {
    .on_conn_init = master_handle_conn_init,
    .on_conn_subscribe = master_handle_conn_subscribe,
    .on_conn_trickle = master_handle_conn_trickle,
    .on_conn_terminate = master_handle_conn_terminate,
    .on_conn_answer = master_handle_conn_answer,
    .on_conn_setup = master_handle_conn_setup,
};

bool master_init_pub_conns(Master* m, int msecs) {
    // init 10 rtc each seconds
    int target = 10 * msecs / 1000;

    if (target > m->ctx.n_pub) {
        target = m->ctx.n_pub;
    }

    while (m->conns.size() < target) {
        int rtc_id = m->conns.size();
        auto conn = new Conn(m->ctx, rtc_id, master_pub_handler, m);
        conn->init();
        m->conns.emplace_back(conn);
    }

    return m->conns.size() == m->ctx.n_pub;
}

bool master_terminate_pub_conns(Master* m) {
    if (m->exit) {
        for (auto& conn : m->conns) {
            if (conn->is_terminating()) {
                continue;
            }
            g_message("master %d: terminate pub rtc %s", getpid(),
                    conn->local_rtc_id().c_str());
            conn->terminate();
        }

        bool all_terminated = true;
        for (auto& conn : m->conns) {
            if (!conn->is_terminated()) {
                all_terminated = false;
                break;
            }
        }

        if (all_terminated) {
            g_message("master %d: all pub rtc terminated", getpid());
            for (auto& conn : m->conns) {
                delete(conn);
            }
            m->conns.clear();
            g_main_loop_quit(m->loop);
            return true;
        }
    }
    return false;
}

bool master_all_pub_setup(Master* m) {
    if (m->conns.size() < m->ctx.n_pub) {
        return false;
    }
    for (auto& conn : m->conns) {
        if (!conn->is_setup()) {
            return false;
        }
    }
    return true;
}

void master_send_packets(Master* m) {
    for (auto& conn : m->conns) {
        if (conn->is_setup() && !conn->is_terminating()) {
            conn->send_packets();
        }
    }
}

void master_spawn_child(Master* m) {
    for (int i = 0; i < m->ctx.workers; ++i) {
        auto worker = subprocess_spawn(i, metric_server_port(m->mserver),
                m->ctx, master_handle_child_exit, m);
        m->workers.emplace(worker);
    }
}

static gboolean master_handle_timeout(gpointer data) {
    auto m = static_cast<Master*>(data);

    auto now = Clock::now();

    auto elapsed = now - m->start_time;

    auto msecs = std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();

    master_init_pub_conns(m, msecs);

    if (master_all_pub_setup(m) && m->workers.empty() && !m->exit) {
        master_spawn_child(m);
    }

    if (master_terminate_pub_conns(m)) {
        g_main_loop_quit(m->loop);
        return G_SOURCE_REMOVE;
    }

    master_send_packets(m);

    return G_SOURCE_CONTINUE;
}

void master_main(Context& ctx, int argc, char* argv[]) {
    auto m = new Master();

    m->ctx = ctx;

    g_message("master start");

    m->mserver = metric_server_open(m->ctx);

    m->loop = g_main_loop_new(nullptr, FALSE);

    auto now = Clock::now();

    m->start_time = now;

    g_unix_signal_add(SIGINT, master_handle_signal, m);

    GSource* timer = g_timeout_source_new(kMasterTimeoutInterval.count());

    g_source_set_callback(timer, master_handle_timeout, m, nullptr);

    g_source_attach(timer, nullptr);

    g_message("master loop start");

    g_main_loop_run(m->loop);

    g_message("master loop stop");

    g_main_loop_unref(m->loop);

    g_source_destroy(timer);

    g_source_unref(timer);

    auto metrics = metric_server_metrics(m->mserver);

    report(argc, argv, metrics);

    if (m->mserver) {
        metric_server_close(m->mserver);
        m->mserver = nullptr;
    }

    delete m;
}
